package com.aiguigu.cn.marketanalysis

import java.sql.Timestamp
import java.util.UUID
import java.util.concurrent.TimeUnit

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.util.Random

/**
  * @author: yangShen
  * @Description: app 分渠道 进行市场分析
  * @Date: 2020/5/6 14:17 
  */
//输入数据样例类
case class MarketingUserBehavior( userId: String, behavior: String, channel: String, timestamp: Long)

//输出结果样例类
case class MarketingViewCount( windowStart: String, windowEnd: String, channel: String, behavior: String, count: Long )

object AppMarketingByChannel {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val dataStream = environment.addSource(new SimulatedEventSource())
      //指定时间戳
      .assignAscendingTimestamps(_.timestamp)
      .filter(_.behavior != "UNINSTALL")
      .map(data =>{
        ( (data.behavior, data.channel), 1L )
      })
      //以渠道和行为类型分组
      //开起窗口，滑动窗口
      .keyBy(_._1)
      .timeWindow(Time.hours(1),Time.seconds(10))
      .process( new MarketingCountByChannel())

    dataStream.print()

    environment.execute("app marketing by channel job")
  }
}

//自定义数据源
class SimulatedEventSource extends RichSourceFunction[MarketingUserBehavior]{

  //定义是否运行的标识位
  var running = true
  //定义用户行为的集合
  val behaviorTypes: Seq[String] = Seq("CLICK", "DOWNLOAD", "INSTALL", "UNINSTALL")
  //定义渠道的集合
  val channelSets: Seq[String] = Seq("wechat", "weibo", "appstore", "huaweistore")
  //定义一个随机发生器
  val rand: Random = new Random()

  override def run(ctx: SourceFunction.SourceContext[MarketingUserBehavior]): Unit = {
    //定义一个生成数据的上线
    val maxElements = Long.MaxValue
    //定义一个计数器
    var count = 0L

    //随机生成所有数据
    while (running && count < maxElements){
      val id = UUID.randomUUID().toString
      val behavior = behaviorTypes(rand.nextInt(behaviorTypes.size))
      val channel = channelSets(rand.nextInt(channelSets.size))
      val ts = System.currentTimeMillis()
      ctx.collect( MarketingUserBehavior( id, behavior, channel, ts) )

      count += 1
      TimeUnit.MILLISECONDS.sleep(10L)  //sleep 10毫秒
    }


  }

  override def cancel(): Unit = running = false
}

//自定义处理函数         --参数(输入in .map的操作, 输出out 自定义MarketingViewCount, key是 .key的操作, TimeWindow)
class MarketingCountByChannel extends ProcessWindowFunction[((String, String), Long), MarketingViewCount, (String, String), TimeWindow]{

  override def process(key: (String, String), context: Context, elements: Iterable[((String, String), Long)], out: Collector[MarketingViewCount]): Unit = {
    val startTs = new Timestamp(context.window.getStart).toString
    val endTs = new Timestamp(context.window.getEnd).toString
    val channel= key._1
    val behavior = key._2
    //此处应该循环elements进行去重： 1.遍历一边塞到set中，2.数据大的话要用布隆过滤
    val count = elements.size
    out.collect(MarketingViewCount(startTs, endTs, channel, behavior, count))
  }
}
